package org.example.realtime.gongdan.log.function;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.asynchttpclient.AsyncHttpClient;
import org.asynchttpclient.DefaultAsyncHttpClient;
import org.asynchttpclient.Response;
import org.example.realtime.gongdan.log.entity.UserLocation;
import org.json.JSONObject;

import java.io.IOException;
import java.sql.Timestamp;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;

public class GaoDeAsyncFunction extends RichAsyncFunction<UserLocation, UserLocation> {

    private final String apiKey;
    private transient AsyncHttpClient asyncHttpClient;

    public GaoDeAsyncFunction(String apiKey) {
        this.apiKey = apiKey;
    }

    @Override
    public void open(Configuration parameters) {
        asyncHttpClient = new DefaultAsyncHttpClient();
    }

    @Override
    public void asyncInvoke(UserLocation input, ResultFuture<UserLocation> resultFuture) {
        String url = String.format(
                "https://restapi.amap.com/v3/geocode/regeo?key=%s&location=%s,%s",
                apiKey, input.getLongitude(), input.getLatitude()
        );

        asyncHttpClient
                .prepareGet(url)
                .execute()
                .toCompletableFuture()
                .thenAccept(response -> handleResponse(response, input, resultFuture))
                .exceptionally(e -> {
                    input.setError("API_ERROR", e.getMessage());
                    resultFuture.complete(Collections.singleton(input));
                    return null;
                });
    }

    private void handleResponse(Response response,
                                UserLocation input,
                                ResultFuture<UserLocation> resultFuture) {
        try {
            JSONObject json = new JSONObject(response.getResponseBody());
            if ("1".equals(json.getString("status"))) {
                JSONObject regeocode = json.getJSONObject("regeocode");
                JSONObject address = regeocode.getJSONObject("addressComponent");

                input.setUpdateTime(new Timestamp(System.currentTimeMillis()));
                input.setProvinceName(address.getString("province"));
                input.setCityName(address.getString("city"));
                input.setDistrictName(address.getString("district"));
                input.setAdcode(address.getString("adcode"));
                input.clearError();
            } else {
                input.setError(json.getString("infocode"), json.getString("info"));
            }
        } catch (Exception e) {
            input.setError("PARSE_ERROR", e.getMessage());
        }
        resultFuture.complete(Collections.singleton(input));
    }

    @Override
    public void close() throws IOException {
        asyncHttpClient.close();
    }
}